package ex55.mm.flow.impound.quote;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import ex55.mm.flow.impound.quote.aggregate.ClientShardInfo;
import ex55.mm.flow.impound.quote.aggregate.ExchangeProduct;
import ex55.mm.flow.impound.quote.common.kit.StringFormat;
import ex55.mm.flow.impound.quote.common.rdsdb.RdsDBHelper;
import ex55.mm.flow.impound.quote.common.rdsdb.ValuesBuilder;
import ex55.mm.flow.impound.quote.exchange.api.*;
import org.jooq.Field;
import org.jooq.Record;
import org.jooq.Result;
import org.jooq.Table;
import org.jooq.impl.DSL;
import org.jooq.impl.SQLDataType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class AggregateDataFacade {
    private final Logger logger = LoggerFactory.getLogger(AggregateDataFacade.class);

    private static class DBT_EX_PROD {
        private static final Table<Record> TABLE = DSL.table("mm_flow_exchange_product");
        private static final Field<String> PRODUCT_ID = DSL.field("product_id", SQLDataType.VARCHAR(191).nullable(false));
        private static final Field<String> EXCHANGE_ID = DSL.field("exchange_id", SQLDataType.VARCHAR(191).nullable(false));
        private static final Field<String> SYMBOL = DSL.field("symbol", SQLDataType.VARCHAR(191).nullable(true));
        private static final Field<String> SYMBOL_ALIAS = DSL.field("symbol_alias", SQLDataType.VARCHAR(191).nullable(true));
        private static final Field<String> STATUS = DSL.field("status", SQLDataType.VARCHAR(191).nullable(false));
        private static final Field<String> BASE_ASSET = DSL.field("base_asset", SQLDataType.VARCHAR(191).nullable(false));
        private static final Field<String> BASE_ASSET_ALIAS = DSL.field("base_asset_alias", SQLDataType.VARCHAR(191).nullable(true));
        private static final Field<String> QUOTE_ASSET = DSL.field("quote_asset", SQLDataType.VARCHAR(191).nullable(false));
        private static final Field<String> QUOTE_ASSET_ALIAS = DSL.field("quote_asset_alias", SQLDataType.VARCHAR(191).nullable(true));
        private static final Field<BigDecimal> TRADE_TAKER_FEE = DSL.field("trade_taker_fee", SQLDataType.DECIMAL(20, 10).nullable(true));
        private static final Field<BigDecimal> WEIGHT = DSL.field("weight", SQLDataType.DECIMAL(20, 10).nullable(false));
        private static final Field<Integer> DEPTH = DSL.field("depth", SQLDataType.INTEGER.nullable(false));
    }

    private static class DBT_EX_PROD_LOG {
        private static final Table<Record> TABLE = DSL.table("mm_flow_exchange_product_log");
        private static final Field<String> LOG_ID = DSL.field("log_id", SQLDataType.VARCHAR(191).nullable(false));
        private static final Field<String> PRODUCT_ID = DSL.field("product_id", SQLDataType.VARCHAR(191).nullable(false));
        private static final Field<String> ACCOUNT_ID = DSL.field("account_id", SQLDataType.VARCHAR(191).nullable(false));
        private static final Field<String> MESSAGE = DSL.field("message", SQLDataType.VARCHAR(200).nullable(true));
    }

    private final Object Data_Lock = new Object(); //用于操作多条数据的时候
    private final ConcurrentHashMap<String, MarketDataDepth> Data_MarketDataDepth = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, ExchangeProduct> Data_ExchangeProduct = new ConcurrentHashMap<>();

    private final RdsDBHelper rdsDBHelper;

    public AggregateDataFacade(RdsDBHelper rdsDBHelper, WebSocketWatchDog webSocketWatchDog) {
        this.rdsDBHelper = rdsDBHelper;
        this.webSocketWatchDog = webSocketWatchDog;
        this.refresh();
        Executors.newSingleThreadScheduledExecutor().scheduleWithFixedDelay(this::refresh, 15, 30, TimeUnit.SECONDS);
    }

    public enum ExStatus {
        ENABLE,
        DISABLE,
    }

    public Map<String, List<ExchangeProduct>> listExchangeProduct(ExchangeType exchangeType, String symbol) {
        Map<String, ExchangeProduct> source = Collections.unmodifiableMap(Data_ExchangeProduct);

        Map<String, List<ExchangeProduct>> exchangeProductMultimap = new HashMap<>();

        source.forEach(
                (s, exchangeProduct) -> {
                    if (exchangeType == null
                            || exchangeType.name().equals(exchangeProduct.getExchangeId())) {
                        if (symbol == null || symbol.equals(exchangeProduct.getSymbol())) {
                            exchangeProductMultimap.computeIfAbsent(
                                    exchangeProduct.getExchangeId(),
                                    s1 -> Lists.newArrayList()).add(exchangeProduct);
                        }
                    }
                });
        return exchangeProductMultimap;
    }

    public void updateWeight(String productId, String accountId, String message, BigDecimal weight) {
        rdsDBHelper.doTransaction(() -> {
            ValuesBuilder prodValues = ValuesBuilder.v()
                    .set(DBT_EX_PROD.WEIGHT, weight);

            int execute1 = rdsDBHelper.masterDslContext()
                    .update(DBT_EX_PROD.TABLE)
                    .set(prodValues.values())
                    .where(DBT_EX_PROD.PRODUCT_ID.eq(productId)).execute();

            if (execute1 != 1) {
                throw new RuntimeException(
                        StringFormat.string("update Error, table:{},values:{},execute:{}"
                                , DBT_EX_PROD.TABLE.getName(), prodValues, execute1)
                );
            }

            ValuesBuilder prodLogValues = ValuesBuilder.v()
                    .set(DBT_EX_PROD_LOG.LOG_ID, ClientShardInfo.generateShortUuid())
                    .set(DBT_EX_PROD_LOG.ACCOUNT_ID, accountId)
                    .set(DBT_EX_PROD_LOG.PRODUCT_ID, productId)
                    .set(DBT_EX_PROD_LOG.MESSAGE, message);

            int execute2 = rdsDBHelper.masterDslContext()
                    .insertInto(DBT_EX_PROD_LOG.TABLE)
                    .set(prodLogValues.values()).execute();

            if (execute2 != 1) {
                throw new RuntimeException(
                        StringFormat.string("insertMustOne Error, table:{},values:{},execute:{}"
                                , DBT_EX_PROD.TABLE.getName(), prodLogValues, execute2)
                );
            }

            this.refresh();
        });
    }

    private class DepthListener implements MessageListener<MarketDataDepth> {

        private final String productId;

        private DepthListener(String productId) {
            this.productId = productId;
        }

        @Override
        public void onMessage(MarketDataDepth message) {
            List<MarketDataDepth.Entry> asks = new ArrayList<>();
            List<MarketDataDepth.Entry> bids = new ArrayList<>();
            if (message.getAsks() != null) {
                for (MarketDataDepth.Entry ask : message.getAsks()) {
                    asks.add(new MarketDataDepth.Entry()
                            .setPrice(ask.getPrice())
                            .setQuantity(ask.getQuantity())
                    );
                }
            }
            if (message.getBids() != null) {
                for (MarketDataDepth.Entry bid : message.getBids()) {
                    bids.add(
                            new MarketDataDepth.Entry()
                                    .setPrice(bid.getPrice())
                                    .setQuantity(bid.getQuantity())
                    );
                }
            }
            Data_MarketDataDepth.put(productId, new MarketDataDepth()
                    .setProduct(message.getProduct())
                    .setUpdatedAt(message.getUpdatedAt())
                    .setRefreshedAt(message.getRefreshedAt())
                    .setAsks(asks)
                    .setBids(bids)
            );
        }

        @Override
        public String onException(ExchangeException e) {
            return null;
        }
    }

    private Map<String, MarketDepthWSConnection> connectionsMap = Maps.newConcurrentMap();

    public void insertMarketDataDepth(String productId, MarketDataDepth marketDataDepth) {
        Data_MarketDataDepth.put(productId, marketDataDepth);
    }

    private final WebSocketWatchDog webSocketWatchDog;

    public synchronized void refresh() {
        Result<Record> exchangeProdRst = rdsDBHelper.consistentReadDSL(true)
                .select()
                .from(DBT_EX_PROD.TABLE)
                .where(DBT_EX_PROD.STATUS.eq(ExStatus.ENABLE.name()))
                .fetch();

        HashMap<String, Record> exProdRstMap = new HashMap<>();
        for (Record record : exchangeProdRst) {
            exProdRstMap.put(record.get(DBT_EX_PROD.PRODUCT_ID), record);
        }

        Set<String> connectedKeys = Sets.newHashSet(connectionsMap.keySet());

        for (Record prodRecord : exchangeProdRst) {
            String prodId = prodRecord.get(DBT_EX_PROD.PRODUCT_ID);
            String exchangeId = prodRecord.get(DBT_EX_PROD.EXCHANGE_ID);
            try {
//                if (!Data_Product.containsKey(prodId)) {
//                    MarketDepthWSConnection connection = new MarketDepthWSConnection(exchangeId, new Product()
//                            .setSymbol(prodRecord.get(DBT_EX_PROD.SYMBOL))
//                            .setBaseAsset(prodRecord.get(DBT_EX_PROD.BASE_ASSET))
//                            .setQuoteAsset(prodRecord.get(DBT_EX_PROD.QUOTE_ASSET))
//                            .setDepth(prodRecord.get(DBT_EX_PROD.DEPTH)), webSocketWatchDog, new DepthListener(prodId));
//                    connection.connect();
//                    connectionsMap.put(prodId, connection);
//                }
//                connectedKeys.remove(prodId);
            } catch (Exception e) {
                logger.warn("refresh error", e);
            }
        }

//        connectedKeys.forEach(key -> {
//            connectionsMap.remove(key).destroy();
//        });

        synchronized (Data_Lock) {
            for (Record r : exProdRstMap.values()) {
                String productId = r.get(DBT_EX_PROD.PRODUCT_ID);
                Data_ExchangeProduct.put(productId, new ExchangeProduct()
                        .setProductId(productId)
                        .setExchangeId(r.get(DBT_EX_PROD.EXCHANGE_ID))
                        .setSymbol(r.get(DBT_EX_PROD.SYMBOL))
                        .setSymbolAlias(r.get(DBT_EX_PROD.SYMBOL_ALIAS))
                        .setBaseAsset(r.get(DBT_EX_PROD.BASE_ASSET))
                        .setBaseAssetAlias(r.get(DBT_EX_PROD.BASE_ASSET_ALIAS))
                        .setQuoteAsset(r.get(DBT_EX_PROD.QUOTE_ASSET))
                        .setQuoteAssetAlias(r.get(DBT_EX_PROD.QUOTE_ASSET_ALIAS))
                        .setTradeTakerFee(r.get(DBT_EX_PROD.TRADE_TAKER_FEE))
                        .setWeight(r.get(DBT_EX_PROD.WEIGHT))
                        .setDepth(r.get(DBT_EX_PROD.DEPTH))
                );
            }
        }
    }

    public Map<String, ExchangeProduct> getData_ExchangeProduct() {
        synchronized (Data_Lock) {
            return new HashMap<>(Data_ExchangeProduct);
        }
    }

    public Map<String, MarketDataDepth> getData_MarketDataDepth() {
        synchronized (Data_Lock) {
            return new HashMap<>(Data_MarketDataDepth);
        }
    }

    public MarketDataDepth getData_MarketDataDepth(String productId) {
        return Data_MarketDataDepth.get(productId);
    }

}
